package org.example.flink.code;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment$;
import org.apache.flink.util.Collector;

import java.util.Arrays;

public class Hello03WordCountByDataStreamUseJava {
    public static void main(String[] args) throws Exception {
        //获取程序运行的环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        //调用Source方法创建DataStream
        DataStreamSource<String> lineStream = environment.socketTextStream("localhost", 19523);
        //调用Transformation(s)
//        SingleOutputStreamOperator<String> wordStream = lineStream.flatMap(new FlatMapFunction<String, String>() {
//            @Override
//            public void flatMap(String line, Collector<String> collector) throws Exception {
//                Arrays.stream(line.split(" ")).forEach(word -> collector.collect(word));
//            }
//        });
//
//        SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream =
//                wordStream.map(new MapFunction<String, Tuple2<String, Integer>>() {
//                    @Override
//                    public Tuple2<String, Integer> map(String w) throws Exception {
//                        return Tuple2.of(w, 1);
//                    }
//                });
//
//        //分区聚合
//        KeyedStream<Tuple2<String, Integer>, String> keyedStream =
//                mapStream.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {
//                    @Override
//                    public String getKey(Tuple2<String, Integer> tp) throws Exception {
//                        return tp.f0;
//                    }
//                });
//
//        SingleOutputStreamOperator<Tuple2<String, Integer>> sumStream = keyedStream.sum("f1");
//        //调用Sink
//        sumStream.print();


        //代码正确--可以推断出具体类型
        lineStream.map(line -> "yjxxt_" + line).print();
        //代码错误--无法进行正确的类型推断
        lineStream.flatMap((line, collector) -> {
            String[] words = line.split(" ");
            for (String word : words) {
                collector.collect(word);
            }
        }).map(word -> Tuple2.of(word, 1)).keyBy(tuple2 ->
                tuple2.f0).sum("f1").print();
        //执行
        environment.execute();
    }
}
